package mapreduce;

import java.util.Vector;

import mapreduce.Record.RecordData;
import mapreduce.Record.RecordPosition;

/**
 * flip the order of record. eg: ((a, b, c), (d, e)) -> ((b, d), (a))
 * 
 * @author foxnono
 * 
 */
public class FlipMapper implements Mapper {
	public FlipMapper(RecordPosition[] newKeySourcePositions,
			RecordPosition[] newValueSourcePositions) {
		m_newKeySourcePositions = newKeySourcePositions;
		m_newValueSourcePositions = newValueSourcePositions;
	}

	private RecordPosition[] m_newKeySourcePositions;
	private RecordPosition[] m_newValueSourcePositions;

	public void map(Record record, Vector<Record> outRecords) {
		RecordData newKeyData = getRecordData(record, m_newKeySourcePositions);
		RecordData newValueData = getRecordData(record,
				m_newValueSourcePositions);
		Record outRecord = Record.getRecord(newKeyData, newValueData);
		outRecords.add(outRecord);
	}

	private RecordData getRecordData(Record record,
			RecordPosition[] sourcePositions) {
		BytesFormat newFormat = null;
		Object newData = null;
		if (sourcePositions.length == 1) {
			newData = record.getEntry(sourcePositions[0]);
			newFormat = record.getFormat(sourcePositions[0]);
		} else {
			newFormat = new MultipleTypeFormat();
			MultipleTypeFormat multipleNewFormat = (MultipleTypeFormat) newFormat;
			newData = new Object[sourcePositions.length];
			Object[] newDataArray = (Object[]) newData;
			for (int i = 0; i < sourcePositions.length; i++) {
				newDataArray[i] = record.getEntry(sourcePositions[i]);
				multipleNewFormat.addFormat(record.getFormat(sourcePositions[i]));
			}
		}
		return new RecordData(newData, newFormat);
	}

	public void release(Vector<Record> outRecords) {
	}

}
